package com.xframe.event.bus.core.msg;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import java.io.IOException;

public class DefaultReceiveConsumer extends DefaultConsumer {

    private IReceiveMsg receiveMsg;

    public DefaultReceiveConsumer(IReceiveMsg receiveMsg) {
        super(receiveMsg.getCurChannel());
        this.receiveMsg = receiveMsg;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body){
        try {
            if(receiveMsg.getMsgProc() != null){
                String message = new String(body, "utf-8");
                receiveMsg.getMsgProc().proc(consumerTag,envelope,properties,message);
            }
            if(!this.receiveMsg.isAutoAck()){
                receiveMsg.getCurChannel().basicAck(envelope.getDeliveryTag(), false);
            }
        } catch (Exception e) {
            if(!this.receiveMsg.isAutoAck()){
                try {
                    receiveMsg.getCurChannel().basicReject(envelope.getDeliveryTag(), true);
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
            e.printStackTrace();
        }
    }
}
